feat: broadcast small build side of SortMergeJoinExec in the static planner - #1904
Conversation
Add ballista.optimizer.broadcast_sort_merge_join_enabled (default false). When enabled, the static distributed planner converts a SortMergeJoinExec whose smaller side is under broadcast_join_threshold_bytes into a CollectLeft hash join and lowers the build side as a broadcast stage, reusing the existing hash-join broadcast path. Redundant input sorts are dropped during conversion. TPC-H SF10 (AQE off): ~16% faster across the suite (q8 1.6x, q11 1.5x, q2 1.4x), identical row counts.
…oadcast-smj # Conflicts: # ballista/scheduler/src/planner.rs
|
AQE does support this, AQE does not support broadcast of build side in SMJ |
milenkovicm
left a comment
There was a problem hiding this comment.
Thanks @andygrove this makes sense.
Just slight correction on my side, I have mistakenly assumed build side of sort merge join has been broadcasted.
I'm not 100% sure but I believe sparks supports broadcasting build side of SMJ, perhaps that could be a good follow up (I think I saw it in TPCDS Q72 SF10)
| planner.plan_query_stages(&job_uuid.to_string().into(), plan, &options)?; | ||
|
|
||
| let mut found_smj = false; | ||
| for stage in &stages { |
There was a problem hiding this comment.
I find it easier to understand tests if they use plan represented as a string, not sure if it could be done here
There was a problem hiding this comment.
Good call — I've rewritten all three SortMergeJoinExec broadcast tests to assert the stage plan as a string via the assert_plan! macro instead of walking the plan tree by hand. The expected plan is now spelled out inline: a broadcast CollectLeft hash join with the sorts stripped when the conversion fires, vs. the unchanged SortMergeJoinExec over sorted inputs when the flag is off or no side is under the threshold. Pushed in 7f868fc.
Enable the SortMergeJoinExec broadcast conversion by default so small build sides are broadcast without opt-in (~16% on TPC-H SF10, AQE off). Disable the flag explicitly in the sort-merge-join client test so it still exercises the plain SortMergeJoinExec execution path.
Rewrite the three SortMergeJoinExec broadcast tests to assert the stage plan as a string via the assert_plan! macro instead of walking the plan tree with manual downcasts, making the expected plan explicit and the tests easier to read.
I filed #1922 to explore this |
…lanner (apache#1904) * feat(scheduler): broadcast small build side of SortMergeJoinExec Add ballista.optimizer.broadcast_sort_merge_join_enabled (default false). When enabled, the static distributed planner converts a SortMergeJoinExec whose smaller side is under broadcast_join_threshold_bytes into a CollectLeft hash join and lowers the build side as a broadcast stage, reusing the existing hash-join broadcast path. Redundant input sorts are dropped during conversion. TPC-H SF10 (AQE off): ~16% faster across the suite (q8 1.6x, q11 1.5x, q2 1.4x), identical row counts. * feat(config): default broadcast_sort_merge_join_enabled to true Enable the SortMergeJoinExec broadcast conversion by default so small build sides are broadcast without opt-in (~16% on TPC-H SF10, AQE off). Disable the flag explicitly in the sort-merge-join client test so it still exercises the plain SortMergeJoinExec execution path. * test: assert SMJ broadcast plans via string snapshots Rewrite the three SortMergeJoinExec broadcast tests to assert the stage plan as a string via the assert_plan! macro instead of walking the plan tree with manual downcasts, making the expected plan explicit and the tests easier to read.
…dopt clobbered 53->54 features
The 54.0.0 merge suffered a delete/modify pathology: files the fork had
deleted relative to base 53.0.0 silently stayed deleted whenever upstream
had not touched them during the 53->54 cycle, and several conflict
resolutions kept stale fork-side file versions. This commit repairs it.
Tree repairs (restored from upstream 54.0.0):
- ballista-cli: tui/domain/{mod,executors}.rs, ui/main/jobs/dot_parser.rs,
exec.rs — the crate did not compile with default features
- examples/standalone-broadcast-join.rs (Cargo.toml still declared it)
- scheduler/src/api/ (routes.rs + handlers.rs + mod.rs) and display.rs
- docs python/ dir, example notebooks, test_jupyter.py, TUI screenshots,
benchmarks/tpch-gen.sh, vendored datafusion*.proto stubs
Re-adopted upstream features the merge clobbered:
- apache#1911 partition pruning (under disable-stage-plan-cache) + tests
- apache#1968 shuffle-read metrics, mapped onto the Spice fetch pipeline, and
child-operator metrics in writer plan displays
- apache#1995 session-keyed runtime cache wired into the executor (plus the
upstream --memory-pool-size FairSpillPool option)
- apache#1982 selective intermediate-stage shuffle cleanup end-to-end (also
applied to the in-memory shuffle manager)
- apache#1949 failed-task REST surfacing + apache#1818 CORS/--disable-rest-api,
TaskManager::get_all_jobs / get_job_config, JobState::get_all_jobs,
ExecutorManager::get_executors_state (upstream names)
- apache#1900/apache#1904 static-planner broadcast promotion (maybe_promote_to_
broadcast, CollectLeft demotion guard, SMJ->hash conversion, broadcast
stage lowering, broadcast_join_threshold_bytes config) + 12 tests
- apache#1999 task duration in finished-task log
- execute_physical_plan client entry point (apache#1924/apache#1941)
- ballista.client.io_retries_times / io_retry_wait_time_ms, wired into
the evict-and-retry fetch loop (fork-preserving defaults: 1 retry, 0ms)
Cleanups and adaptations:
- deleted dead executor/src/client_pool.rs, orphaned AQE optimizer files,
committed .pending-snap artifacts
- shuffle.remote_read_prefer_flight now defaults to true: the block-IO
transport cannot serve sort-based shuffle (default-on), so the old
default pair was incoherent; removed the block-IO sort-shuffle test
cases accordingly
- regenerated TPC-H plan-stability goldens (TopK single-stage merge from
the fork's #28.1 was not reflected in the goldens taken from upstream)
- SPICE_FORK_CHANGES.md updated to match reality (#61/#62a rows, apache#1951
disposition, repair-commit inventory)
Which issue does this PR close?
Closes #1679.
Rationale for this change
The static
DefaultDistributedPlannercan broadcast a small build side(
HashJoinExec(Partitioned)→CollectLeft, lowered as a broadcast shuffle),but only for
HashJoinExec. Ballista defaultsprefer_hash_join = false, soTPC-H joins are planned as
SortMergeJoinExecand broadcast never fires on thedefault (non-AQE) path: small dimension tables get a full hash shuffle and force
the large side to reshuffle on the join key.
A
SortMergeJoinExechas noCollectLeftmode, so when a side is small enoughthis PR converts it to a
HashJoinExec(CollectLeft)(the build side fits inmemory by definition, so the no-spill reason Ballista prefers sort-merge does not
apply) and reuses the existing broadcast lowering. This mirrors what the AQE
resolver already does for sort-merge joins.
What changes are included in this PR?
ballista.optimizer.broadcast_sort_merge_join_enabled(default
false).maybe_promote_to_broadcastconverts a small-sideSortMergeJoinExecto aHashJoinExec(Partitioned)(dropping the now-redundant inputSortExec) andruns it through the existing threshold/swap/promotion path. Handles the
ProjectionExecthatswap_inputsinserts to preserve column order.CollectLeftwith sortsstripped; SMJ left unchanged when the flag is off or no side is under the
threshold.
Are there any user-facing changes?
Yes — a new opt-in config. With it enabled, a
SortMergeJoinExecwhose buildside fits under
broadcast_join_threshold_bytesis executed as a broadcast(
CollectLeft) hash join instead of two hash-partitioned shuffles. No SQLsemantics change.
TPC-H SF10 (AQE off, 8 partitions): ~16% faster across the 22-query suite
(q8 1.6×, q11 1.5×, q2 1.4×, q17 1.4×, q9 1.3×), with identical row counts.
A follow-up can also strip the join-key
RepartitionExecfrom the convertedinputs (build broadcast from natural partitions, probe kept in place) for a
further gain; the existing hash-join broadcast path shares that limitation.